// $Id: jpeg.c 100 2011-10-21 15:05:36Z cedric.shih@gmail.com $
/*
 * Copyright 2011 Cedric Shih (cedric dot shih at gmail dot com)
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation; either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/tcp.h>

#include <zcbuffer.h>

#include "log.h"
#include "util.h"

#include "jpeg.h"

#define DEFAULT_MAX_QTABLES 4
#define MIN(a,b) a == 0 ? b : (a > b ? b : a)
#define MAX(a,b) a > b ? a : b

int niohttp_jpeg_init(struct niohttp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame);

int niohttp_jpeg_make_header(struct niohttp_jpeg *transport);

int niohttp_jpeg_start(struct niohttp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame);

int niohttp_jpeg_write_header(struct niohttp_jpeg *transport);

int niohttp_jpeg_write_qtable(struct niohttp_jpeg *transport);

int niohttp_jpeg_write_payload(struct niohttp_jpeg *transport);

ssize_t niohttp_jpeg_clear_buffers(struct zcqueue *queue);

int niohttp_jpeg_qos(struct niohttp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame);

void niohttp_jpeg_stat_cb(int fd, short event, void *arg);

struct niohttp_transport niohttp_jpeg = {
		niohttp_jpeg_create,
		niohttp_jpeg_destroy,
		niohttp_jpeg_setup,
		niohttp_jpeg_teardown,
		niohttp_jpeg_send,
		niohttp_jpeg_pending,
		niohttp_jpeg_next,
		niohttp_jpeg_stat,
};

void *niohttp_jpeg_create(struct event_base *evbase)
{
	int rc;
	struct niohttp_jpeg *transport;

	if (!(transport = calloc(1, sizeof(*transport)))) {
		rc = errno ? errno : -1;
		niohttp_error("calloc: %s", strerror(rc));
		goto finally;
	}

	transport->evbase = evbase;

	rc = 0;

finally:
	if (rc != 0) {
		niohttp_jpeg_destroy(transport);
		transport = NULL;
		errno = rc;
	}
	return transport;
}

void niohttp_jpeg_destroy(void *obj)
{
	struct niohttp_jpeg *transport = obj;

	if (!transport) {
		return;
	}

	if (transport->setup) {
		niohttp_jpeg_teardown(transport);
	}

	free(transport);
}

int niohttp_jpeg_setup(void *obj, int fd, struct niohttp_transport_param *param)
{
	int rc;
	struct niohttp_jpeg *transport = obj;

	if (!transport || !param) {
		rc = EINVAL;
		niohttp_error("null pointer(s)");
		goto finally;
	}

	if (transport->setup) {
		rc = EPERM;
		niohttp_error("already set up");
		goto finally;
	}

	niohttp_debug("setting up: %s:%d", inet_ntoa(param->remote.sin_addr),
			ntohs(param->remote.sin_port));

	if (!(transport->type = strdup(param->type))) {
		rc = errno ? errno : -1;
		niohttp_error("strdup: %s", strerror(rc));
		goto finally;
	}

	if (!(transport->boundary = strdup(param->boundary))) {
		rc = errno ? errno : -1;
		niohttp_error("strdup: %s", strerror(rc));
		goto finally;
	}

	if (!(transport->header = evbuffer_new())) {
		rc = errno ? errno : -1;
		niohttp_error("evbuffer_new: %s", strerror(rc));
		goto finally;
	}

	if (!(transport->queue = zcqueue_new())) {
		rc = errno ? errno : -1;
		niohttp_error("zcqueue_new: %s", strerror(rc));
		goto finally;
	}

	if (!(transport->dropping = zcqueue_new())) {
		rc = errno ? errno : -1;
		niohttp_error("zcqueue_new: %s", strerror(rc));
		goto finally;
	}

	evtimer_set(&transport->ev, niohttp_jpeg_stat_cb, transport);

	if (event_base_set(transport->evbase, &transport->ev)) {
		rc = errno ? errno : -1;
		niohttp_error("failed to set event base: %s", strerror(rc));
		goto finally;
	}

	transport->tv.tv_sec = 1;
	transport->tv.tv_usec = 0;

	if (event_add(&transport->ev, &transport->tv)) {
		rc = errno ? errno : -1;
		niohttp_error("failed to add timer: %s", strerror(rc));
		goto finally;
	}

	transport->fd = fd;
	memcpy(&transport->param, param, sizeof(transport->param));
	transport->setup = 1;

	rc = 0;

finally:
	return rc;
}

int niohttp_jpeg_teardown(void *obj)
{
	int rc;
	ssize_t n;
	struct niohttp_jpeg *transport = obj;

	if (!transport) {
		rc = EINVAL;
		niohttp_error("null pointer(s)");
		goto finally;
	}

	if (!transport->setup) {
		rc = EPERM;
		niohttp_error("not yet set up");
		goto finally;
	}

	if (event_initialized(&transport->ev) &&
			evtimer_pending(&transport->ev, NULL)) {
		if ((rc = evtimer_del(&transport->ev))) {
			niohttp_error("failed to delete timer: %s",
					strerror(rc));
			goto finally;
		}
	}

	if (transport->buffer) {
		zcbuffer_free(transport->buffer);
		transport->buffer = NULL;
	}

	if (transport->dropping) {
		if ((n = niohttp_jpeg_clear_buffers(
				transport->dropping)) < 0) {
			rc = errno ? errno : -1;
			niohttp_error("failed to clear buffers: %s",
					strerror(rc));
			goto finally;
		}

		zcqueue_free(transport->dropping);
		transport->dropping = NULL;
	}

	if (transport->queue) {
		if ((n = niohttp_jpeg_clear_buffers(
				transport->queue)) < 0) {
			rc = errno ? errno : -1;
			niohttp_error("failed to clear buffers: %s",
					strerror(rc));
			goto finally;
		}

		zcqueue_free(transport->queue);
		transport->queue = NULL;
	}

	if (transport->header) {
		evbuffer_free(transport->header);
		transport->header = NULL;
	}

	if (transport->boundary) {
		free(transport->boundary);
		transport->boundary = NULL;
	}

	if (transport->type) {
		free(transport->type);
		transport->type = NULL;
	}

	transport->setup = 0;
	rc = 0;

finally:
	return rc;
}

int niohttp_jpeg_stat(void *obj, struct niohttp_transport_stat *stat)
{
	int rc;
	struct niohttp_jpeg *transport = obj;

	if (!transport || !stat) {
		rc = EINVAL;
		niohttp_error("null pointer(s)");
		goto finally;
	}

	if (!transport->setup) {
		rc = EPERM;
		niohttp_error("not yet set up");
		goto finally;
	}

	memcpy(stat, &transport->old, sizeof(*stat));
	rc = 0;

finally:
	return rc;
}

int niohttp_jpeg_send(void *obj, const struct timeval *timestamp,
		struct zcbuffer *frame)
{
	int rc;
	struct niohttp_jpeg *transport = obj;

	if (!transport || !timestamp || !frame) {
		rc = EINVAL;
		niohttp_error("null pointer(s)");
		goto finally;
	}

	if (!transport->setup) {
		rc = EPERM;
		niohttp_error("not yet set up");
		goto finally;
	}

	transport->new.incoming.units++;
	transport->new.incoming.bytes += zcbuffer_remaining(frame);

	niohttp_trace("sending frame %p to: %s:%d", frame,
			inet_ntoa(transport->param.remote.sin_addr),
			ntohs(transport->param.remote.sin_port));

	if ((rc = niohttp_jpeg_qos(transport, timestamp, frame))) {
		niohttp_error("failed to apply QoS rules: %s", strerror(rc));
		goto finally;
	}

	if ((rc = niohttp_jpeg_start(transport,
			timestamp, frame))) {
		niohttp_error("failed to send frame: %s",
				strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	return rc;
}

int niohttp_jpeg_pending(void *obj)
{
	int pending;
	struct niohttp_jpeg *transport = obj;

	if (!transport) {
		niohttp_error("null pointer");
		return 0;
	}

	if ((pending = (transport->buffer || (transport->queue &&
			zcqueue_size(transport->queue) > 0)))) {
		niohttp_trace("pending");
	} else {
		niohttp_trace("not pending");
	}

	return pending;
}

int niohttp_jpeg_next(void *obj)
{
	int rc;
	struct niohttp_jpeg *transport = obj;
	struct zcbuffer *buffer = NULL;
	struct timeval timestamp;

	if (!transport) {
		rc = EINVAL;
		niohttp_error("null pointer");
		goto finally;
	}

	if (!transport->buffer) {
		niohttp_trace("processing next frame");

		if (!(buffer = zcqueue_take(transport->queue,
				&timestamp))) {
			rc = errno ? errno : -1;
			niohttp_error("failed to take: %s", strerror(rc));
			goto finally;
		}

		if ((rc = niohttp_jpeg_init(transport, &timestamp,
				buffer))) {
			niohttp_error("failed to init: %s", strerror(rc));
			goto finally;
		}

		if ((rc = niohttp_jpeg_make_header(transport))) {
			niohttp_error("failed to make header: %s",
					strerror(rc));
			goto finally;
		}
	}

	niohttp_trace("continuing JPEG frame: %p", transport->buffer);

	if (EVBUFFER_LENGTH(transport->header) > 0 &&
			(rc = niohttp_jpeg_write_header(transport))) {
		if (rc != EAGAIN && rc != EINPROGRESS) {
			niohttp_error("failed to write header: %s",
					strerror(rc));
		}
		goto finally;
	}

	if ((rc = niohttp_jpeg_write_payload(transport))) {
		if (rc != EAGAIN && rc != EINPROGRESS) {
			niohttp_error("failed to write payload: %s",
					strerror(rc));
		}
		goto finally;
	}

	rc = 0;

finally:
	if (rc != 0) {
		if (buffer) {
			zcbuffer_free(buffer);
			transport->buffer = NULL;
		}
	}
	return rc;
}

int niohttp_jpeg_init(struct niohttp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *buffer)
{
	int rc;
	struct timeval now, diff;
	unsigned int duration;

	if (memcmp(&transport->timestamp, timestamp,
			sizeof(transport->timestamp))) {
		if ((rc = gettimeofday(&now, NULL)) == -1) {
			rc = errno ? errno : -1;
			niohttp_error("failed to get time: %s", strerror(rc));
			goto finally;
		}

		if (transport->start.tv_sec != 0 &&
				transport->start.tv_usec != 0) {
			if ((rc = niohttp_timeval_diff(&transport->start,
					&now, &diff))) {
				niohttp_error("failed to diff time: %s",
						strerror(rc));

				memset(&transport->start, '\0',
						sizeof(transport->start));
			} else {
				duration = diff.tv_sec * 1000000 + diff.tv_usec;
				transport->new.duration.min = MIN(
						transport->new.duration.min,
						duration);
				transport->new.duration.max = MAX(
						transport->new.duration.max,
						duration);
			}
		}

		memcpy(&transport->start, &now, sizeof(transport->start));
	}

	memcpy(&transport->timestamp, timestamp, sizeof(transport->timestamp));
	transport->buffer = buffer;
	rc = 0;

finally:
	return rc;
}

int niohttp_jpeg_make_header(struct niohttp_jpeg *transport)
{
	int rc;

	niohttp_trace("making header: %s", transport->type);

	evbuffer_drain(transport->header, EVBUFFER_LENGTH(transport->header));

	if (evbuffer_add_printf(transport->header,
			"--%s\r\n"
			"Content-Type: %s\r\n"
			"Content-Length: %d\r\n"
			"X-Timestamp: %ld.%06ld\r\n"
			"\r\n",
			transport->boundary,
			transport->type,
			zcbuffer_remaining(transport->buffer),
			transport->timestamp.tv_sec,
			transport->timestamp.tv_usec) <= 0) {
		rc = errno ? errno : -1;
		niohttp_error("failed to prepare header: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	return rc;
}

int niohttp_jpeg_qos(struct niohttp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame)
{
	int rc;
	ssize_t n;
	size_t dropped, queued;

	dropped = 0;

	if (transport->buffer) {
		niohttp_trace("clearing frame(s) newer than %ld.%06ld",
				transport->timestamp.tv_sec,
				transport->timestamp.tv_usec);

		if ((rc = zcqueue_take_newer(transport->queue,
				transport->dropping, &transport->timestamp))) {
			niohttp_error("failed to take newer: %s", strerror(rc));
			goto finally;
		}

		dropped = zcqueue_size(transport->dropping);

		if ((n = niohttp_jpeg_clear_buffers(
				transport->dropping)) < 0) {
			rc = errno ? errno : -1;
			niohttp_error("failed to clear: %s", strerror(rc));
			goto finally;
		}

		transport->new.dropped.units += dropped;
		transport->new.dropped.bytes += n;
	}

	if (dropped > 0) {
		queued = zcqueue_size(transport->queue);
		
		niohttp_warn("qos applied %s:%d => dropped=%d, queued=%d",
				inet_ntoa(transport->param.remote.sin_addr),
				ntohs(transport->param.remote.sin_port),
				dropped, queued);
	}

	rc = 0;

finally:
	return rc;
}

int niohttp_jpeg_start(struct niohttp_jpeg *transport,
		const struct timeval *timestamp, struct zcbuffer *frame)
{
	int rc;
	struct zcbuffer *clone = NULL;

	niohttp_trace("streaming frame %p: %s:%d", frame,
			inet_ntoa(transport->param.remote.sin_addr),
			ntohs(transport->param.remote.sin_port));

	if ((rc = niohttp_jpeg_qos(transport, timestamp, frame))) {
		niohttp_error("failed to apply QoS rules: %s", strerror(rc));
		goto finally;
	}

	if (!(clone = zcbuffer_clone(frame))) {
		rc = errno ? errno : -1;
		niohttp_error("failed to clone: %s",
				strerror(rc));
		goto finally;
	}

	if ((rc = zcqueue_put(transport->queue, timestamp, clone))) {
		niohttp_error("failed to put: %s", strerror(rc));
		goto finally;
	}

	rc = 0;

finally:
	if (rc != 0) {
		if (clone) {
			zcbuffer_free(clone);
		}
	}
	return rc;
}

int niohttp_jpeg_write_header(struct niohttp_jpeg *transport)
{
	int rc;
	ssize_t n;
	void *data;
	size_t len;

	niohttp_trace("writing header: %d bytes",
			EVBUFFER_LENGTH(transport->header));

	data = EVBUFFER_DATA(transport->header);
	len = EVBUFFER_LENGTH(transport->header);

	if ((n = write(transport->fd, data, len)) <= 0) {
		rc = errno ? errno : -1;

		if (rc != EAGAIN && rc != EINPROGRESS) {
			niohttp_error("failed to write: %s", strerror(rc));
		}

		goto finally;
	}

	niohttp_trace("header written: %d bytes", n);

	evbuffer_drain(transport->header, n);
	transport->new.http.bytes += n;
	rc = 0;

finally:
	return rc;
}

int niohttp_jpeg_write_payload(struct niohttp_jpeg *transport)
{
	int rc;
	ssize_t n;

	if (zcbuffer_remaining(transport->buffer) <= 0) {
		rc = ENODATA;
		niohttp_error("buffer already written");
		goto finally;
	}

	niohttp_trace("writing payload: %d/%d bytes",
			zcbuffer_remaining(transport->buffer),
			zcbuffer_limit(transport->buffer));

	if ((n = zcbuffer_write(transport->buffer, transport->fd)) <= 0) {
		rc = errno ? errno : -1;

		if (rc != EAGAIN && rc != EINPROGRESS) {
			niohttp_error("failed to write: %s", strerror(rc));
		}

		goto finally;
	}

	niohttp_trace("payload written: %d bytes", n);

	transport->new.http.bytes += n;

	if (zcbuffer_remaining(transport->buffer) <= 0) {
		niohttp_trace("frame transport completed: %p",
				transport->buffer);

		transport->new.outgoing.units++;
		transport->new.outgoing.bytes +=
				zcbuffer_position(transport->buffer);

		zcbuffer_free(transport->buffer);
		transport->buffer = NULL;

		if (memcmp(&transport->previous, &transport->timestamp,
				sizeof(transport->previous))) {
			transport->new.instances++;
			memcpy(&transport->previous, &transport->timestamp,
					sizeof(transport->previous));
		}
	}

	rc = 0;

finally:
	return rc;
}

ssize_t niohttp_jpeg_clear_buffers(struct zcqueue *queue)
{
	int rc;
	struct zcbuffer *buffer;
	ssize_t size = 0;

	while (zcqueue_size(queue) > 0) {
		if (!(buffer = zcqueue_take(queue, NULL))) {
			rc = errno ? errno : -1;
			niohttp_error("failed to take: %s", strerror(rc));
			goto finally;
		}
		size += zcbuffer_remaining(buffer);
		zcbuffer_free(buffer);
	}

	rc = 0;

finally:
	if (rc != 0) {
		size = -1;
		errno = rc;
	}
	return size;
}

void niohttp_jpeg_stat_cb(int fd, short event, void *arg)
{
	struct niohttp_jpeg *transport = arg;

	memcpy(&transport->old, &transport->new, sizeof(transport->old));
	memset(&transport->new, '\0', sizeof(transport->new));
	event_add(&transport->ev, &transport->tv);
}
